package com.ikas.ai.framework.flow;

import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.ikas.ai.framework.core.event.FlowEvent;
import com.ikas.ai.framework.core.listener.FlowListener;
import com.ikas.ai.framework.core.listener.FlowLogListener;
import com.ikas.ai.framework.node.ChainNode;
import lombok.extern.slf4j.Slf4j;

import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.stream.Collectors;

/**
 * @Description: 评分流程
 * @Author: qi.yilai
 * @Date: 2023/1/12 19:48
 * @Version: 1.0.0
 **/
@Slf4j
public class ScoreFlow {

    // 流程id
    private Long flowId;

    // 流程实例id
    private Long flowInstanceId;

    // 流程名称
    private String name;

    private long sceneId;

    private String sceneName;

    // 所有节点
    private List<ChainNode> nodes = Lists.newArrayList();

    // 开始节点
    private List<ChainNode> starts = Lists.newArrayList();

    private volatile long startTime;

    private volatile long endTime;

    private volatile FlowStatus status = FlowStatus.CREATED;

    // 下一步需要执行的节点
    protected LinkedList<ChainNode> currentNodes = Lists.newLinkedList();

    // 处理失败的节点
    protected List<ChainNode> errors = Lists.newArrayList();

    // 正在执行的节点
    protected List<ChainNode> doingNodes = Lists.newArrayList();

    // 监听器
    protected List<FlowListener> listeners = Lists.newArrayList();

    // 正在异步执行的任务
    private Map<ChainNode, Future> doingFutures = Maps.newIdentityHashMap();

    // 全局变量
    private Map<String, Object> globalMap = Maps.newHashMap();

    public ScoreFlow(String name, long flowId, long flowInstanceId, long sceneId, String sceneName) {
        this.name = name;
        this.flowId = flowId;
        this.flowInstanceId = flowInstanceId;
        this.sceneId = sceneId;
        this.sceneName = sceneName;
        listeners.add(new FlowLogListener());
    }

    public ScoreFlow(String name, long flowId, long flowInstanceId, long sceneId, String sceneName, Map<String, Object> globalMap) {
        this(name, flowId, flowInstanceId, sceneId, sceneName);
        this.globalMap = globalMap;
    }

    public boolean start(ExecutorService executorService) {
        setStatus(FlowStatus.STARTING);
        startTime = System.currentTimeMillis();
        FlowEvent flowEvent = createFlowEvent(FlowStatus.STARTING, new JSONObject());
        fireEvent(flowEvent);

        log.info("{}开始节点数量{}", getName(), starts.size());
        for (ChainNode chainNode : starts) {
            executeNode(executorService, chainNode);
        }
        return true;
    }

    private void executeNode(ExecutorService executorService, ChainNode chainNode) {
        Future future = executorService.submit(() -> {
            addDoingNode(chainNode);
            boolean flag = chainNode.execute();
            finishChainNode(chainNode, flag);
            if (isFinish() || isFatal()) {
                finish();
                return;
            }
            if (flag) {
                execute(executorService);
            }
        });
        doingFutures.put(chainNode, future);
    }

    public synchronized void execute(ExecutorService executorService) {
        List<ChainNode> filters = this.currentNodes.stream().filter(ChainNode::canExecute).collect(Collectors.toList());

        for (ChainNode chainNode : filters) {
            log.info("chainName: {}", chainNode.getCurrentNode().name());
            currentNodes.remove(chainNode);
            executeNode(executorService, chainNode);
        }
    }

    public void finish() {
        endTime = System.currentTimeMillis();
        if (errors.isEmpty()) {
            setStatus(FlowStatus.SUCCESS);
            FlowEvent flowEvent = createFlowEvent(FlowStatus.SUCCESS, new JSONObject());
            fireEvent(flowEvent);
        } else {
            setStatus(FlowStatus.FAILURE);
            FlowEvent flowEvent = createFlowEvent(FlowStatus.FAILURE, new JSONObject());
            fireEvent(flowEvent);
        }
    }

    /**
     * 流程已经结束
     *
     * @return
     */
    public boolean isOver() {
        return status == FlowStatus.STOP || status == FlowStatus.SUCCESS || status == FlowStatus.FAILURE;
    }

    /**
     * 流程停止
     */
    public synchronized void stop() {
        setStatus(FlowStatus.STOP);
        endTime = System.currentTimeMillis();
        String cause = "用户终止流程";
        for (ChainNode chainNode : doingNodes) {
            Future future = doingFutures.get(chainNode);
            if (Objects.nonNull(future)) {
                future.cancel(true);
                finishChainNode(chainNode, false);
            }
        }
        FlowEvent flowEvent = createFlowEvent(FlowStatus.STOP, cause);
        fireEvent(flowEvent);
    }

    public synchronized void finishChainNode(ChainNode doneNode, boolean success) {
        // 检查，避免已经调用过 finishChainNode
        if (!doingNodes.contains(doneNode)) {
            return;
        }
        removeDoingNode(doneNode);
        if (success) {
            doneNode.fillPreviousResult();
            doneNode.setFinished(true);
            appendCurrentNodes(doneNode.getNext());
        } else {
            addFailChainNode(doneNode);
            doneNode.markFatal();
        }
        if (doingFutures.containsKey(doneNode)) {
            doingFutures.remove(doneNode);
        }
    }

    public synchronized boolean isFinish() {
        boolean executing = currentNodes.isEmpty() && doingNodes.isEmpty();
        if (!executing) {
            return false;
        }
        return true;
    }

    /**
     * 严重故障，无法继续执行
     *
     * @return
     */
    public synchronized boolean isFatal() {
        return doingNodes.isEmpty() && (currentNodes.isEmpty() || currentFatal());
    }

    private boolean currentFatal() {
        for (ChainNode chainNode : currentNodes) {
            if (!chainNode.isFatal()) {
                return false;
            }
        }
        return true;
    }

    private FlowEvent createFlowEvent(FlowStatus status, Object target) {
        FlowEvent flowEvent = new FlowEvent();
        flowEvent.setTarget(target);
        flowEvent.setStatus(status);
        flowEvent.setNow(System.currentTimeMillis());
        flowEvent.setOwner(this);
        return flowEvent;
    }

    private void fireEvent(FlowEvent flowEvent) {
        listeners.forEach(flowListener -> {
            try {
                flowListener.fireEvent(flowEvent);
            } catch (Exception exception) {
                log.error("flow fire event error", exception);
            }
        });
    }

    private synchronized void addFailChainNode(ChainNode chainNode) {
        if (!errors.contains(chainNode)) {
            errors.add(chainNode);
        }
    }

    public synchronized boolean appendCurrentNodes(List<ChainNode> chainNodes) {
        for (ChainNode chainNode : chainNodes) {
            if (!currentNodes.contains(chainNode)) {
                currentNodes.add(chainNode);
            }
        }
        return true;
    }

    public FlowStatus getStatus() {
        return status;
    }

    public void setStatus(FlowStatus status) {
        this.status = status;
    }

    public void addListener(FlowListener flowListener) {
        if (!listeners.contains(flowListener)) {
            listeners.add(flowListener);
        }
    }

    private synchronized void addDoingNode(ChainNode chainNode) {
        doingNodes.add(chainNode);
    }

    private synchronized void removeDoingNode(ChainNode chainNode) {
        doingNodes.remove(chainNode);
    }

    public Long getFlowId() {
        return flowId;
    }

    public Long getFlowInstanceId() {
        return flowInstanceId;
    }

    public String getName() {
        return name;
    }

    public long getStartTime() {
        return startTime;
    }

    public long getEndTime() {
        return endTime;
    }

    public int errorSize() {
        return errors.size();
    }

    public void setNodes(List<ChainNode> nodes) {
        this.nodes = nodes;
    }

    public void setStarts(List<ChainNode> starts) {
        this.starts = starts;
    }

    public Map<String, Object> getGlobalMap() {
        return globalMap;
    }

    public void setGlobalMap(Map<String, Object> globalMap) {
        this.globalMap = globalMap;
    }

    public long getSceneId() {
        return sceneId;
    }

    public void setSceneId(long sceneId) {
        this.sceneId = sceneId;
    }

    public List<ChainNode> getNodes() {
        return nodes;
    }

    public String getSceneName() {
        return sceneName;
    }
}
